package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.hmdp.constant.MessageConstant;
import com.hmdp.dto.Result;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.ISeckillVoucherService;
import com.hmdp.service.IVoucherOrderService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.service.IVoucherService;
import com.hmdp.utils.RedisConstants;
import com.hmdp.utils.RedisIdWorker;
import com.hmdp.utils.UserHolder;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p>
 *  服务实现类
 * </p>
 */
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {
    @Autowired
    private IVoucherService voucherService;

    @Autowired
    private ISeckillVoucherService seckillVoucherService;

    @Autowired
    private IVoucherOrderService voucherOrderService;

    @Autowired
    private RedisIdWorker redisIdWorker;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Autowired
    private RedissonClient redissonClient;

    // 定义代理对象的成员变量
    IVoucherOrderService proxy;

    // 定义lua脚本的静态代码块，启动时自动执行
    private static final DefaultRedisScript<Long> SECKILL_SCRIPT;
    static {
        SECKILL_SCRIPT = new DefaultRedisScript<Long>();
        // SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/seckill.lua"));
        SECKILL_SCRIPT.setLocation(new ClassPathResource("lua/stream_mq_seckill.lua"));
        SECKILL_SCRIPT.setResultType(Long.class);
    }

    // 创建单线程线程池
    private final static ExecutorService SECKILL_ORDER_EXECUTOR = Executors.newSingleThreadExecutor();


    /**
     * 当前类初始化完毕就立马执行该方法
     */
    @PostConstruct
    private void init() {
        // 执行线程任务
        SECKILL_ORDER_EXECUTOR.submit(new VoucherOrderServiceImpl.VoucherOrderHandler());
    }

    // 队列名
    String queueName = "stream.orders";

    /**
     * 线程任务: 不断从消息队列中获取订单
     */
    private class VoucherOrderHandler implements Runnable {

        @Override
        public void run() {
            while (true){
                try {
                    // 获取当前处理的优惠券的id

                    // 1.从消息队列中获取信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                    List<MapRecord<String, Object, Object>> mapRecordList = stringRedisTemplate.opsForStream().read(
                            Consumer.from("g1", "c1"),
                            StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create(queueName , ReadOffset.lastConsumed())
                    );
                    // 2.判断消息是否获取成功
                    if(mapRecordList == null || mapRecordList.isEmpty()){
                        // 2.1失败就进行下一次循环
                        continue;
                    }
                    // 3.对消息进行解析
                    MapRecord<String, Object, Object> record = mapRecordList.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 4，获取成功就进行下单操作
                    handleVoucherOrder(voucherOrder);
                    // 5.进行ACK确认 - 处理pendingList的消息
                    // 5.1获取当前拿到的消息的id
                    RecordId messageId = record.getId();
                    stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", messageId);
                } catch (Exception e) {
                    log.error("订单处理异常", e);
                    handlePendingList();
                }
            }
        }
    }

    /**
     * 处理pending-List的消息
     */
    private void handlePendingList() {
        while (true){
            try {
                // 1.从pending-List中获取信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
                List<MapRecord<String, Object, Object>> mapRecordList = stringRedisTemplate.opsForStream().read(
                        Consumer.from("g1", "c1"),
                        StreamReadOptions.empty().count(1),
                        // 0 表示从pending-List中读取第一条
                        StreamOffset.create(queueName, ReadOffset.from("0"))
                );
                // 2.判断消息是否获取成功
                if(mapRecordList == null || mapRecordList.isEmpty()){
                    // 2.1失败就结束循环
                    break;
                }
                // 3.对消息进行解析
                MapRecord<String, Object, Object> record = mapRecordList.get(0);
                Map<Object, Object> value = record.getValue();
                VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                // 4，获取成功就进行下单操作
                handleVoucherOrder(voucherOrder);
                // 5.进行ACK确认 - 处理pendingList的消息
                // 5.1获取当前拿到的消息的id
                RecordId messageId = record.getId();
                stringRedisTemplate.opsForStream().acknowledge(queueName, "g1", messageId);
            } catch (Exception e) {
                log.error("订单pending-List处理异常", e);
                // 线程休眠20毫秒
                try {
                    Thread.sleep(20);
                } catch (InterruptedException ex) {
                    throw new RuntimeException(ex);
                }
            }
        }
    }


    /**
    // 创建存储订单的阻塞队列
    // capacity - 预定义的阻塞队列的长度
    private BlockingQueue<VoucherOrder> orderTasks = new ArrayBlockingQueue<>(1024 * 1024);
    /**
     * 线程任务: 不断从阻塞队列中获取订单
    private class VoucherOrderHandler implements Runnable {
        @Override
        public void run() {
            while (true){
                // 从阻塞队列中获取订单
                try {
                    VoucherOrder voucherOrder = orderTasks.take();
                    // 创建订单
                    handleVoucherOrder(voucherOrder);
                } catch (Exception e) {
                    log.error("订单状态异常", e);
                }
            }
        }
    }*/

    /**
     * 创建订单，基于消息队列异步创建
     * @param voucherOrder
     */
    private void handleVoucherOrder(VoucherOrder voucherOrder) {
        // 获取当前用户id
        Long userId = voucherOrder.getUserId();
        // 获取分布式锁对象
        // 基于redisson获取锁对象
        RLock lock = redissonClient.getLock(RedisConstants.SECKILL_VOUCHER_KEY + userId);

        boolean isLock = lock.tryLock();
        if(!isLock){
            // 获取锁对象失败，返回错误信息，或者重试
            log.error("获取锁对象失败");
            return;
        }
        try {
            // 获取当前事务对象的代理对象，创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效
            proxy.createVoucherOrder(voucherOrder);
        } catch (Exception e) {
            log.error("订单创建失败", e);
        } finally {
            // 释放锁
            lock.unlock();
        }
    }

    /**
     * 创建订单 - 异步交互的创建订单（开启单独线程）
     *
     * @param voucherOrder
     */
    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        // 获取当前登陆用户id
        Long userId = voucherOrder.getId();
        // 获取当前订单id
        Long voucherId = voucherOrder.getVoucherId();
        // 一人一单，判断该用户是否已经拥有该订单
        Integer count = voucherOrderService.query().eq("user_id", userId).eq("voucher_id", voucherId).count();
        if(count > 0){
            log.error("一人只能下一单");
        }
        // 扣减库存
        // 基于乐观锁解决的超卖问题
        boolean success = seckillVoucherService.update().
                setSql("stock = stock - 1").
                eq("voucher_id", voucherId).
                gt("stock", 0).
                update();
        if (!success) {
            throw new RuntimeException("秒杀券扣减失败");
        }
        // 保存订单
        boolean flag = voucherOrderService.save(voucherOrder);
        if (!flag){
            throw new RuntimeException("创建秒杀券订单失败");
        }

    }

    /**
     * 抢购秒杀优惠券 基于lua脚本和redis-stream消息队列实现秒杀优化
     * @param voucherId
     * @return
     */
    @Override
    public Result buySeckillVoucher(Long voucherId) {
        // 获取当前用户id
        Long userId = UserHolder.getUser().getId();
        // 订单id
        Long voucherOrderId = redisIdWorker.nextId(RedisConstants.SECKILL_VOUCHER_KEY);
        // 1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString(), voucherOrderId.toString()
        );
        // 2.判断结果是否为0
        int flag = result.intValue();
        if(flag != 0){
            // 3.非0直接报错返回异常
            return Result.error(flag == 1 ? MessageConstant.STOCK_IS_NULL : MessageConstant.USER_ALREADY_BUY_VOUCHER);
        }

        // 3.获取当前事务对象的代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 4.返回订单id
        return Result.success(voucherOrderId);
    }

    /**
     * 抢购秒杀优惠券 基于lua脚本和阻塞队列实现异步秒杀
     * @param voucherId
     * @return
     */
    /**@Override
    public Result buySeckillVoucher(Long voucherId) {
        // 获取当前用户id
        Long userId = UserHolder.getUser().getId();
        // 1.执行lua脚本
        Long result = stringRedisTemplate.execute(
                SECKILL_SCRIPT,
                Collections.emptyList(),
                voucherId.toString(), userId.toString()
        );
        // 2.判断结果是否为0
        int flag = result.intValue();
        if(flag != 0){
            // 3.非0直接报错返回异常
            return Result.error(flag == 1 ? MessageConstant.STOCK_IS_NULL : MessageConstant.USER_ALREADY_BUY_VOUCHER);
        }
        // 4.为0则将订单id，用户id存入阻塞队列中，实现异步下单
        // 4.1创建订单信息
        // 创建订单
        VoucherOrder voucherOrder = new VoucherOrder();
        // 用户id
        voucherOrder.setUserId(userId);
        // 优惠券id
        voucherOrder.setVoucherId(voucherId);
        // 订单id
        Long voucherOrderId = redisIdWorker.nextId(RedisConstants.SECKILL_VOUCHER_KEY);
        voucherOrder.setId(voucherOrderId);
        // 4.2将订单存入到阻塞队列内部
        orderTasks.add(voucherOrder);
        // 5.获取当前事务对象的代理对象
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        // 6.返回订单id
        return Result.success(voucherOrderId);
    }*/

    /**
     * 抢购秒杀优惠券
     * @param voucherId
     * @return
     */
    /**
     @Override
     public Result buySeckillVoucher(Long voucherId) {
     //获取优惠券信息
     SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
     //判断秒杀是否开始
     LocalDateTime beginTime = voucher.getBeginTime();
     if(beginTime.isAfter(LocalDateTime.now())){
     return Result.error(MessageConstant.SECKILL_IS_NOT_BEGIN);
     }
     //判断秒杀是否结束
     LocalDateTime endTime = voucher.getEndTime();
     if(endTime.isBefore(LocalDateTime.now())){
     return Result.error(MessageConstant.SECKILL_IS_END);
     }
     //判断库存是否充足
     Integer stock = voucher.getStock();
     if(stock < 1){
     return Result.error(MessageConstant.STOCK_IS_NULL);
     }
     //获取当前登陆用户id
     Long userId = UserHolder.getUser().getId();
     //通过userId添加悲观锁，intern是将用户id字符串放进字符串常量中，保证每一次获取的字符串是唯一的
     //         synchronized (userId.toString().intern()){
     //            获取当前事务对象的代理对象，创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效
     //            ISeckillVoucherService proxy = (ISeckillVoucherService) AopContext.currentProxy();
     //            return proxy.createVoucherOrder(voucherId);
     //        }

     // 通过redis实现的分布式锁方案解决集群模式下一人一单问题
     //        SimpleRedisLock lock = new SimpleRedisLock(
     //                RedisConstants.SECKILL_VOUCHER_KEY + userId, stringRedisTemplate);

     // 获取分布式锁对象
     // 基于redisson获取锁对象
     RLock lock = redissonClient.getLock(RedisConstants.SECKILL_VOUCHER_KEY + userId);

     boolean isLock = lock.tryLock();
     if(!isLock){
     // 获取锁对象失败，返回错误信息，或者重试
     return Result.error(MessageConstant.USER_ALREADY_BUY_VOUCHER);
     }
     // 获取当前事务对象的代理对象，创建代理对象，使用代理对象调用第三方事务方法， 防止事务失效
     try {
     ISeckillVoucherService proxy = (ISeckillVoucherService) AopContext.currentProxy();
     return proxy.createVoucherOrder(voucherId);
     } catch (IllegalStateException e) {
     throw new RuntimeException(e);
     } finally {
     // 释放锁
     lock.unlock();
     }
     }*/

    /**
     * 创建订单
     * @param voucherId
     * @return
     */
    /**
     @Transactional
     public Result createVoucherOrder(Long voucherId) {
     //获取当前登陆用户id
     Long userId = UserHolder.getUser().getId();
     //一人一单，判断该用户是否已经拥有该订单
     Integer count = voucherOrderService.query().eq("user_id", userId).eq("voucher_id", voucherId).count();
     if(count > 0){
     return Result.error(MessageConstant.USER_ALREADY_BUY_VOUCHER);
     }
     //扣减库存
     // 基于乐观锁解决的超卖问题
     boolean success = this.update().
     setSql("stock = stock - 1").
     eq("voucher_id", voucherId).
     gt("stock", 0).
     update();
     if (!success) {
     return Result.error(MessageConstant.STOCK_IS_NULL);
     }
     //创建订单
     VoucherOrder voucherOrder = new VoucherOrder();
     //用户id
     voucherOrder.setUserId(userId);
     //优惠券id
     voucherOrder.setVoucherId(voucherId);
     //订单id
     Long voucherOrderId = redisIdWorker.nextId(RedisConstants.SECKILL_VOUCHER_KEY);
     voucherOrder.setId(voucherOrderId);
     //保存订单
     boolean flag = voucherOrderService.save(voucherOrder);
     if (!flag){
     return Result.error(MessageConstant.ERROR_TO_CREATE_VOUCHER_ORDER);
     }
     //返回订单id
     return Result.success(voucherOrderId);
     }*/
}
